package com.wisc.application;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

import org.apache.commons.io.FileUtils;
import org.apache.log4j.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.protobuf.ServiceException;

import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.consumer.Consumer;
import edu.wisc.hevea.exception.BrokerUnavailableException;
import edu.wisc.hevea.messaging.Message;

public class ConsumerApplication {

	private static final String CONSUMER_PATH_PREFIX = "/home/anand/Documents/DistributedSystems/HeveaLibraries/consumer/";
	private static final Logger logger = LoggerFactory
			.getLogger(ConsumerApplication.class);

	/**
	 * @param args
	 */
	public static void main(String[] args) {

		if (args.length < 1) {
			System.err.println(" Please enter the consumer's host name!");
			return;
		}

		org.apache.log4j.BasicConfigurator.configure();
		org.apache.log4j.Logger.getRootLogger().setLevel(Level.INFO);

		String appID = "APP_ID_1";
		short replication = 2;
		List<String> dcsHosted = Arrays.asList(new String[] {
				ApplicationConstants.VIRGINIA_DC, ApplicationConstants.IAD_DC,
				ApplicationConstants.DUBLIN_DC });
		ApplicationPolicy appPolicy = new ApplicationPolicy(appID, replication,
				dcsHosted);

		String hostname = args[0]; // ApplicationConstants.VIRGINIA_DC;
		Consumer consumer = new Consumer(hostname, appPolicy);

		String topic1 = "TOPIC_1";

		while (true) {
			try {
				Message m = consumer.getMessage(topic1);
				processMessage(m, hostname);
				Thread.sleep(3 * 1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (BrokerUnavailableException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (ServiceException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

	}

	private static void processMessage(Message m, String hostName)
			throws IOException {
		if (m.getMsgId().equals("-1")) {
			logger.debug("No pending msgs from broker!");
			return;
		}
		logger.debug("Consuming message: " + m.getMsgId());
		File consumerWorkFile = new File(CONSUMER_PATH_PREFIX + hostName
				+ ".worklog");
		FileUtils.write(consumerWorkFile,
				"Processed content: " + m.getContent(), true);
	}

}
